package org.zjt.http.okhttp;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.schedulers.Schedulers;
import lombok.extern.slf4j.Slf4j;
import okhttp3.*;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.CountDownLatch;

/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-05-29 下午5:15
 * @see
 */
@Slf4j
public class Main {
    private static final OkHttpClient client = new OkHttpClient();


    public static void main(String[] args) throws Exception {
        reactiveAsyncHttp();
//        asyncHttp();
    }


    /**
     * 异步发送http
     *
     * @throws Exception
     */
    public static void asyncHttp() throws Exception {

        Request request = new Request.Builder().url("http://www.baidu.com").build();

        LocalDateTime start = LocalDateTime.now();

        CountDownLatch countDownLatch = new CountDownLatch(10000);

        for (int i = 0; i < 10000; i++) {
            client.newCall(request).enqueue(new Callback() {

                @Override
                public void onFailure(Call call, IOException e) {
                    e.printStackTrace();
                    countDownLatch.countDown();
                }

                @Override
                public void onResponse(Call call, Response response) throws IOException {
                    countDownLatch.countDown();
                    try (ResponseBody responseBody = response.body()) {
                        if (!response.isSuccessful()) throw new IOException("Unexpected code " + response);

                        Headers responseHeaders = response.headers();
                        for (int i = 0, size = responseHeaders.size(); i < size; i++) {
                            System.out.println(responseHeaders.name(i) + ": " + responseHeaders.value(i));
                        }

                        System.out.println(responseBody.string());
                    }
                }
            });
        }

        countDownLatch.await();
        long between = ChronoUnit.MILLIS.between(start, LocalDateTime.now());
        log.info("asyncHttp 耗时:{} ms", between);
    }


    public static void syncHttp() throws Exception {

        Request request = new Request.Builder().url("https://www.baidu.com").build();

        LocalDateTime start = LocalDateTime.now();
        for (int j = 0; j < 10000; j++) {
            try (Response response = client.newCall(request).execute()) {
                if (!response.isSuccessful()) throw new IOException("Unexpected code " + response);

                Headers responseHeaders = response.headers();
                for (int i = 0; i < responseHeaders.size(); i++) {
                    System.out.println(responseHeaders.name(i) + ": " + responseHeaders.value(i));
                }

                System.out.println(response.body().string());
            }
        }
        log.info("耗时:{} ms", ChronoUnit.MILLIS.between(start, LocalDateTime.now()));
    }


    public static void reactiveAsyncHttp() throws Exception {
        Request request = new Request.Builder().url("http://www.baidu.com").build();

        LocalDateTime start = LocalDateTime.now();

        CountDownLatch countDownLatch = new CountDownLatch(10000);

        Flowable.range(1, 10000).flatMap(i -> Flowable.create(
                flowableEmitter ->

                        client.newCall(request).enqueue(new Callback() {

                            @Override
                            public void onFailure(Call call, IOException e) {
                                e.printStackTrace();
                                flowableEmitter.onError(e);
                                flowableEmitter.onComplete();
                                countDownLatch.countDown();
                            }

                            @Override
                            public void onResponse(Call call, Response response) throws IOException {
                                countDownLatch.countDown();
                                try (ResponseBody responseBody = response.body()) {
                                    if (!response.isSuccessful()) throw new IOException("Unexpected code " + response);

                                    Headers responseHeaders = response.headers();
                                    for (int i = 0, size = responseHeaders.size(); i < size; i++) {
                                        System.out.println(responseHeaders.name(i) + ": " + responseHeaders.value(i));
                                    }
                                    String content = responseBody.string();
                                    flowableEmitter.onNext(content);
                                    flowableEmitter.onComplete();
                                }
                            }
                        }), BackpressureStrategy.BUFFER)
        ).subscribeOn(Schedulers.io()).subscribe(e -> System.out.println("   " + e));


        countDownLatch.await();
        long between = ChronoUnit.MILLIS.between(start, LocalDateTime.now());
        log.info("reactiveAsyncHttp 耗时:{} ms", between);
    }
}
